package com.zhny.test;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.parquet.Strings;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.linalg.distributed.RowMatrix;

//PCA主成分分析
public class PCAUtil {
    public static void exc(String dataFilePath, String resultFilePath) {
        SparkConf conf = new SparkConf().setAppName("PCA Algorithm").setMaster("local[*]");
        conf.set("spark.driver.allowMultipleContexts", "true");

        SparkContext sc = new SparkContext(conf);
        JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);

        JavaRDD<Vector> rows = null;

        if (Strings.isNullOrEmpty(dataFilePath)) {
            List<Vector> data = Arrays.asList(
                    Vectors.sparse(5, new int[] {1, 3}, new double[] {1.0, 7.0}),
                    Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),
                    Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)
            );

            rows = jsc.parallelize(data);
        } else {
            JavaRDD<String> data = jsc.textFile(dataFilePath);

            rows = data.map(new Function<String, Vector>() {
                private static final long serialVersionUID = 1L;

                @Override
                public Vector call(String line) throws Exception {
                    String[] elements = line.split(",");

                    double[] vectors = new double[elements.length];

                    for (int i = 0; i < elements.length; i++) {
                        vectors[i] = Double.parseDouble(elements[i]);
                    }

                    return Vectors.dense(vectors);
                }
            }).cache();
        }

        // Create a RowMatrix from JavaRDD<Vector>.
        RowMatrix mat = new RowMatrix(rows.rdd());

        // Compute the top 4 principal components.
        // Principal components are stored in a local dense matrix.
        Matrix pc = mat.computePrincipalComponents(4);

        // Project the rows to the linear space spanned by the top 4 principal components.
        RowMatrix projected = mat.multiply(pc);
        // $example off$
        Vector[] collectPartitions = (Vector[]) projected.rows().collect();
        System.out.println("Projected vector of principal component:");

        FileWriter fos = null;
        try {
            fos = new FileWriter(new File(resultFilePath));

            for (Vector vector : collectPartitions) {
                fos.write("\t" + vector);
            }

            fos.flush();
            fos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

        jsc.stop();
    }
}
